package udf.matching;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.LinkedList;

import javax.swing.text.html.HTML.Tag;

import model.Collocation;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

public class MergeTags extends EvalFunc<DataBag> {

	private boolean initialized = false;

	private BagFactory bagFactory = BagFactory.getInstance();

	private Hashtable<String, Collocation> hyphenized = new Hashtable<String, Collocation>();

	// stores AB --> A_B
	private Hashtable<String, String> simple = new Hashtable<String, String>();

	/**
	 * Tuple fields
	 * 
	 * 0 set of tags of the bookmark 1 set of tags hpyhenized (dictionary)
	 * 
	 * 
	 * @param tuple
	 * @throws IOException
	 */

	/**
	 * Position 0 : a_b Position 1 : freq,
	 * 
	 * threshold_pmi="1.0E-5" threshold_ratio="0.015" a_b, freq, pmi, ratio
	 * threshold_pmi="1.0E-5" threshold_ratio="0.015"
	 * 
	 * 
	 * 
	 * mergeTags($1, hyphen_dict.$1, $threshold, $threshold_ratio,
	 * $threshold_pmi)
	 * 
	 * 
	 */

	public void init(Tuple tuple) throws IOException {

		DataBag bag = (DataBag) tuple.get(1);

		Iterator<Tuple> iterator = bag.iterator();

		while (iterator.hasNext()) {

			Tuple tup = iterator.next();

			String tag = (String) tup.get(0);
			Integer f = (Integer) tup.get(1);

			Float pmi = (Float) tup.get(2);
			Float ratio = (Float) tup.get(3);

			Collocation col = new Collocation();
			col.freq_h = f;
			col.pmi = pmi;
			col.ratio = ratio;

			// System.out.println("Adding: " + tag + "\t" + temp[0]);
			hyphenized.put(tag, col);
			// System.out.println("Init dictionary:" + tag + "\t"+ freq);
			// add simple
			String ab = tag.replace("_", "");
			simple.put(ab, tag);

		}

		initialized = true;
	}

	public LinkedList<String> getSetofTags(Tuple tuple) throws IOException {

		LinkedList<String> list = new LinkedList<String>();

		DataBag bag = (DataBag) tuple.get(0);

		Iterator<Tuple> iterator = bag.iterator();
		// System.out.println("Printing input of tags");
		while (iterator.hasNext()) {

			Tuple tup = iterator.next();

			// for (int i = 0; i < tup.size(); i++) {

			// System.out.print(tup.get(i) + "\t");
			//
			// }
			// System.out.println(tup.size());
			list.add((String) tup.get(3));

		}

		return list;
	}

	/**
	 * 
	 * 
	 * 
	 * Verifies if tag is of the form AB where A_B is in dictionary
	 * 
	 * it returns the equivalent A_B in dictionary if any if it does not exists
	 * if returns the tag
	 * 
	 * 
	 * only returns A_B if ratio is greater than threshold and PMI greater than
	 * thresold (for now than 0)
	 * 
	 * 
	 * threshold_pmi="1.0E-5" threshold_ratio="0.015"
	 * 5.8317525100382587E-8 4707696.0 5.947706490031113E-6
	 * 
	 * 
	 * @param tag
	 * @return
	 */
	public String splitInsideBigram(String tag, int threshold,
			float threshold_ratio, float threshold_pmi) {

		String ab = tag;
		if (simple.containsKey(ab)) {
			System.out
					.println(hyphenized.get(simple.get(ab)).ratio + "\t"
							+ hyphenized.get(simple.get(ab)).ratio + "\t"
							+ threshold_ratio + "\t"
							+ hyphenized.get(simple.get(ab)).pmi + "\t"
							+ threshold_pmi);

		}

		if (simple.containsKey(ab)
				&& hyphenized.get(simple.get(ab)).freq_h >= threshold
				&& (hyphenized.get(simple.get(ab)).ratio >= threshold_ratio || hyphenized
						.get(simple.get(ab)).pmi > threshold_pmi)) {

			// System.out.println("Entro:" +ab );

			String a_b = simple.get(ab); 

			return simple.get(ab);
		}

		return null;

	}

	/**
	 * join pair of tags
	 * 
	 * tagA tagB exists as A_B with a frequency greater or equal than threshold
	 * 
	 */

	public String mergePairOfTags(String a, String b, int threshold, float pmi) {

		String a_b = a + "_" + b;
		String b_a = b + "_" + a;

		int f_ab = 0;
		int f_ba = 0;

		if (hyphenized.containsKey(a_b)) {
			f_ab = hyphenized.get(a_b).freq_h;

		}

		if (hyphenized.containsKey(b_a)) {
			f_ba = hyphenized.get(b_a).freq_h;

		}

		if (f_ab == f_ba && f_ab == 0) {
			return null;
		}

		String key = "";
		int freq = 0;

		if (f_ab >= f_ba) {
			key = a_b;
			freq = f_ab;

		} else {

			key = b_a;
			freq = f_ab;
		}

		if (freq >= threshold) {
			return key;

		}

		return null;

	}

	/*
	 * This funtion take a list of tags from the same bookmark and:
	 * 
	 * tries all combinations to find bigrams of the form A_B from the
	 * dictionary
	 * 
	 * For now i will merge everything
	 * 
	 * tag A can be associated to many others, but if it was associated to at
	 * least another tag it won't be included as such into the final set
	 * 
	 * 
	 * POSSIBLE IMPROVEMENT : Analyze cases in which leaving independent tags in
	 * the set is convenient, based on PMI metrics
	 */

	public HashSet<String> mergeTags(LinkedList<String> tags, int threshold,
			float threshold_pmi) {

		HashSet<String> n = new HashSet<String>();
		HashSet<String> added = new HashSet<String>();

		for (int i = 0; i < tags.size() - 1; i++) {
			String a = tags.get(i);
			
			
			
			//for each tag check the case in which the tags is of the form a_b
			//decided if we have to separated
			if(hyphenized.containsKey(a)){
				
				float pmi = hyphenized.get(a).pmi;
				
				if(pmi<threshold_pmi && pmi>0.0){
					
					String ab[]= a.split("_");
					n.add(ab[0]);
					n.add(ab[1]);
					
				}
				
			}
			

			for (int j = i + 1; j < tags.size(); j++) {
				String b = tags.get(j);
				String a_b = mergePairOfTags(a, b, threshold, threshold_pmi);

				// not marge anything
				if (a_b != null) {
					n.add(a_b);

					Collocation col = hyphenized.get(a_b);

					if (col.pmi < threshold_pmi && col.pmi >= 0.0f) {
						added.add(a);
						added.add(b);
					}

				}
			} // end internal if j

		} // end outer if i

		// go throught the list and add tags that were not merge

		for (int i = 0; i < tags.size(); i++) {

			if (!added.contains(tags.get(i))) {
				n.add(tags.get(i));

			}
		}

		added.clear();
		added = null;
		return n;
	}

	public LinkedList<String> convertToHyphenized(LinkedList<String> tags,
			int threshold, float threshold_ratio, float threshold_pmi) {

		LinkedList<String> set = new LinkedList<String>();

		Iterator<String> iter = tags.iterator();

		while (iter.hasNext()) {

			String tag = iter.next();

			String h = splitInsideBigram(tag, threshold, threshold_ratio,
					threshold_pmi);
			if (h == null) {
				set.add(tag);

			} else {

				// now we verified if we have the case
				// legostarwars --> lego_starwars --> lego_star_wars

				/*
				 * now we are adding separete tags in cases such as
				 * 
				 * lego_star_wars , this will create the tags lego, star_wars,
				 * lego_starwars
				 */

				String ab[] = h.split("_");

				String h_a = splitInsideBigram(ab[0], threshold,
						threshold_ratio, threshold_pmi);
				String h_b = splitInsideBigram(ab[1], threshold,
						threshold_ratio, threshold_pmi);

				String n_a = null;
				String n_b = null;

				if (h_a != null && h_b == null) {

					n_a = h_a;
					if (!set.contains(n_a)) {
						set.add(n_a);

					}

					if (!set.contains(ab[1])) {
						set.add(ab[1]);

					}

					set.add(h_a + "_" + ab[1]);

				} else if (h_b != null && h_a == null) {

					n_b = h_b;
					if (!set.contains(n_b)) {
						set.add(n_b);

					}

					if (!set.contains(ab[0])) {
						set.add(ab[0]);

					}

					set.add(ab[0] + "_" + n_b);

				} else if (h_a != null && h_b != null) {

					n_b = h_b;
					if (!set.contains(n_b)) {
						set.add(n_b);

					}

					n_a = h_a;
					if (!set.contains(n_a)) {
						set.add(n_a);

					}
					set.add(n_a + "_" + n_b);

				} else if (h_a == null && h_b == null) {

					set.add(h);

					// decide if ab[0], ab[1] should be included separetely

					float pmi = hyphenized.get(simple.get(tag)).ratio;
					if (pmi <= threshold_pmi && pmi > 0) {

						if (!set.contains(ab[0])) {
							set.add(ab[0]);

						}

						if (!set.contains(ab[1])) {
							set.add(ab[1]);

						}

					}

				}

			}

		}
		return set;

	}

	/**
	 * 
	 * decides in the tags a, b from a_b should be added separetely
	 * 
	 * @param a_b
	 * @return
	 */

	/**
	 * Input tuple
	 * 
	 * 0 tags
	 * 
	 * 1 dictionary of bigrams a_b
	 * 
	 * 2 freq threshold
	 * 
	 * 
	 */

	public DataBag exec(Tuple input) throws IOException {

		if (!initialized)
			init(input);

		LinkedList<String> list = getSetofTags(input);
		
		// System.out.println("Converting:");
		// for (int i = 0; i < list.size(); i++) {

		// System.out.print(list.get(i) + "\t");

		// }
		// System.out.println("");

		// convert tags of the form AB to A_B

		int threshold = Integer.valueOf((String) input.get(2));
		
		float threshold_ratio = Float.valueOf((String) input.get(3));
		float threshold_pmi = Float.valueOf((String) input.get(4));
		
		LinkedList<String> listA = convertToHyphenized(list, threshold,
				threshold_ratio, threshold_pmi);

		// merge bigrams A B to A_B

		HashSet<String> listB = mergeTags(listA, threshold, threshold_pmi);

		DataBag output = bagFactory.newDefaultBag();
		
		// generate output

		Iterator<String> iter = listB.iterator();
		// System.out.println("Result:");
		while (iter.hasNext()) {

			String tag = iter.next();
			Tuple t = TupleFactory.getInstance().newTuple();
			// System.out.print(tag + "\t");
			t.append(tag);
			output.add(t);

		}
		// System.out.println("");
		// debud

		return output;

	}

}
